package cn.com.itg.function.process;

import cn.com.itg.common.BuildVOUtil;
import cn.com.itg.pojo.domain.entity.ZMSEG;
import cn.com.itg.pojo.domain.vo.MatdocSendVO;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.FlinkException;

import java.io.Serializable;
import java.util.Iterator;
import java.util.Map;

@Slf4j
public class MatdocCoZmsegCPFunction extends KeyedCoProcessFunction<Tuple3<String, String, String>, MatdocSendVO, ZMSEG, MatdocSendVO> implements Serializable {

//    private KeyedStateStore

    private final long lowerBound = Time.days(-4).toMilliseconds() + 1L;

    private final long upperBound = Time.seconds(1).toMilliseconds() - 1L;

    private static final String LEFT_BUFFER = "LEFT_BUFFER";
    private static final String RIGHT_BUFFER = "RIGHT_BUFFER";

    private transient MapState<Long, MatdocSendVO> leftBuffer;
    private transient MapState<Long, ZMSEG> rightBuffer;

    private transient long leftCurrentWatermark;

    private transient long rightCurrentWatermark;

    private transient long currentWatermark;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        MapStateDescriptor<Long, MatdocSendVO> left_state = new MapStateDescriptor<>(
                LEFT_BUFFER,
                TypeInformation.of(new TypeHint<Long>() {
                }),
                TypeInformation.of(new TypeHint<MatdocSendVO>() {
                })
        );
        left_state.enableTimeToLive(
                StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.milliseconds(upperBound > 1 ? upperBound : 1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                        .build()
        );

        MapStateDescriptor<Long, ZMSEG> right_state = new MapStateDescriptor<>(
                RIGHT_BUFFER,
                TypeInformation.of(new TypeHint<Long>() {
                }),
                TypeInformation.of(new TypeHint<ZMSEG>() {
                })
        );
        right_state.enableTimeToLive(
                StateTtlConfig
                        .newBuilder(org.apache.flink.api.common.time.Time.milliseconds(
                                (upperBound - lowerBound) > 1 ? (upperBound - lowerBound) : 1)
                        )
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                        .build()
        );

        this.leftBuffer = getRuntimeContext().getMapState(left_state);
        this.rightBuffer = getRuntimeContext().getMapState(right_state);
    }


    @Override
    public void processElement1(MatdocSendVO matdocSendVO, Context context, Collector<MatdocSendVO> collector) throws Exception {
        log.info("matdocSendVO by ztmm 【{}】",JSON.toJSONString(matdocSendVO));
        if (this.leftCurrentWatermark == Long.MIN_VALUE ){
            this.leftCurrentWatermark = context.timestamp();
        }else if( context.timestamp() != Long.MIN_VALUE){
            this.leftCurrentWatermark = Math.max(this.leftCurrentWatermark,context.timestamp());
        }
        this.processElement(matdocSendVO, context, collector, this.leftBuffer, this.rightBuffer, this.lowerBound, this.upperBound, true);
    }

    @Override
    public void processElement2(ZMSEG zmseg, Context context, Collector<MatdocSendVO> collector) throws Exception {
//        log.info("zms");
        if (this.rightCurrentWatermark == Long.MIN_VALUE ){
            this.rightCurrentWatermark = context.timestamp();
        }else if( context.timestamp() != Long.MIN_VALUE){
            this.rightCurrentWatermark = Math.max(this.rightCurrentWatermark,context.timestamp());
        }
        this.processElement(zmseg, context, collector, this.rightBuffer, this.leftBuffer, -this.upperBound, -this.lowerBound, false);

    }

    private <THIS, OTHER> void processElement(THIS ourValue, Context ourContext, Collector<MatdocSendVO> collector, MapState<Long, THIS> ourBuffer, MapState<Long, OTHER> otherBuffer, long relativeLowerBound, long relativeUpperBound, boolean isLeft) throws Exception {
        this.currentWatermark = Math.min(this.leftCurrentWatermark,this.rightCurrentWatermark);
        System.out.println(ourContext.getCurrentKey());
        long ourTimestamp = ourContext.timestamp();
        if (ourTimestamp == Long.MIN_VALUE) {
            throw new FlinkException("Long.MIN_VALUE timestamp: Elements used in interval stream joins need to have timestamps meaningful timestamps.");
        } else
            if (!this.isLate(ourTimestamp))
            {
            addToBuffer(ourBuffer, ourValue, ourTimestamp);
            Iterator var12 = otherBuffer.entries().iterator();

            while (true) {
                Map.Entry bucket;
                long timestamp;
                do {
                    do {
                        if (!var12.hasNext()) {
                            long cleanupTime = relativeUpperBound > 0L ? ourTimestamp + relativeUpperBound : ourTimestamp;
                            if (isLeft) {
//                                cleanupTime = relativeUpperBound > 0L ? ourContext.timerService().currentProcessingTime() + relativeUpperBound : ourTimestamp;
                                ourContext.timerService().registerProcessingTimeTimer(cleanupTime);
                            } else {
                                ourContext.timerService().registerEventTimeTimer(cleanupTime);
                            }
                            return;
                        }

                        //另一条流的信息
                        bucket = (Map.Entry) var12.next();
                        //另一条流的时间戳
                        timestamp = (Long) bucket.getKey();
                    } while (timestamp < ourTimestamp + relativeLowerBound);
                } while (timestamp > ourTimestamp + relativeUpperBound);

                OTHER entry = (OTHER) bucket.getValue();
                if (isLeft) {
                    log.info("is left merge");
                    log.info("left join ZMSEG 【{}】", JSON.toJSONString(entry));
                    collector.collect(BuildVOUtil.buildSendVOByZmseg((MatdocSendVO) ourValue, (ZMSEG) entry));
                } else {
                    log.info("is right merge");
                    log.info("ZMSEG 【{}】", JSON.toJSONString(ourValue));
                    collector.collect(BuildVOUtil.buildSendVOByZmseg((MatdocSendVO) entry, (ZMSEG) ourValue));
                }
            }
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<MatdocSendVO> out) throws Exception {

        super.onTimer(timestamp, ctx, out);
        long left_timestamp = this.upperBound <= 0L ? timestamp : timestamp - this.upperBound;
        long right_timestamp = this.lowerBound <= 0L ? timestamp + this.lowerBound : timestamp;
        //如果左侧流存在
        if (this.leftBuffer.contains(left_timestamp)) {
            //将该流直接输出
            MatdocSendVO matdocSendVO = this.leftBuffer.get(left_timestamp);
            log.info("on timer no zmseg 【{}】",JSON.toJSONString(matdocSendVO));
            out.collect(matdocSendVO);
            this.leftBuffer.remove(left_timestamp);
        }
        //如果在右侧流存在
        if (this.rightBuffer.contains(right_timestamp)) {
            //将该流直接从state中移除
            this.leftBuffer.remove(right_timestamp);
        }
    }

    private boolean isLate(long timestamp) {
        long currentWatermark = this.currentWatermark;
        return currentWatermark != -9223372036854775808L && timestamp < currentWatermark;
    }

    private static <T> void addToBuffer(MapState<Long, T> buffer, T value, long timestamp) throws Exception {
        buffer.put(timestamp, value);
    }
}
